package es

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/jaegertracing/jaeger/model"
	"github.com/olivere/elastic/v7"
	trace2 "gluenet/pkg/ops/impl/es/trace"
	"hash/fnv"
	"strings"
	"time"
)

type Es struct {
	Client *elastic.Client
	Bulk   *elastic.BulkProcessor
	Domain trace2.FromDomain
}

func NewEs(ctx context.Context, url string) (*Es, error) {
	options := []elastic.ClientOptionFunc{elastic.SetURL(url),
		elastic.SetSniff(false), elastic.SetHealthcheck(true),
	}
	rd, err := elastic.NewClient(options...)
	if err != nil {
		return nil, err
	}
	_, err = rd.IndexPutTemplate("jaeger-span").BodyString(trace2.SpanMap).Do(ctx)
	if err != nil {
		return nil, err
	}
	_, err = rd.IndexPutTemplate("jaeger-service").BodyString(trace2.ServiceMap).Do(ctx)
	if err != nil {
		return nil, err
	}

	recorder, err := rd.BulkProcessor().BulkSize(5 * 1000 * 1000).
		Workers(1).BulkActions(1000).
		FlushInterval(time.Millisecond * 200).Do(ctx)
	if err != nil {
		return nil, err
	}

	d := trace2.NewFromDomain(false, []string{}, "@")

	return &Es{
		Client: rd,
		Bulk:   recorder,
		Domain: d,
	}, nil
}

func (r *Es) WriteSpan(data []byte) error {
	span := &model.Span{}
	//fmt.Printf("span data: %s\n", data)
	if err := json.Unmarshal(data, span); err != nil {
		return fmt.Errorf("unmarshall json span err %v", err)
	}
	//if err := jsonpb.Unmarshal(bytes.NewReader(data), span); err != nil {
	//	return fmt.Errorf("unmarshall span err %v", err)
	//}
	traceElk := r.Domain.FromDomainEmbedProcess(span)
	service := trace2.Service{
		ServiceName:   traceElk.Process.ServiceName,
		OperationName: traceElk.OperationName,
	}

	h := fnv.New64a()
	_, _ = h.Write([]byte(service.ServiceName))
	_, _ = h.Write([]byte(service.OperationName))
	id := fmt.Sprintf("%x", h.Sum64())

	date := time.Unix(0, int64(traceElk.StartTimeMillis*1e6)).Format("2006-01-02")

	bulkIndexReq1 := elastic.NewBulkIndexRequest()
	bulkIndexReq1.Index("jaeger-service-" + date).Id(id).Doc(service)

	bulkIndexReq2 := elastic.NewBulkIndexRequest()
	bulkIndexReq2.Index("jaeger-span-" + date).Doc(&traceElk)

	r.Bulk.Add(bulkIndexReq1)
	r.Bulk.Add(bulkIndexReq2)

	return nil
}

func (r *Es) WriteLog(ctx context.Context, key string, data interface{}) error {
	if _, err := r.Client.Index().Index(strings.ToLower(key)).BodyJson(data).Do(ctx); err != nil {
		return err
	}
	return nil
}
